package com.hliushi.spark.rdd

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}
import org.junit.Test

import scala.collection.mutable.ListBuffer

/**
 * descriptions:
 *
 * author: Hliushi
 * date: 2021/5/14 10:11
 */
class TransformationOp {

  val conf: SparkConf = new SparkConf().setMaster("local[6]").setAppName("transformation_op")
  val sc = new SparkContext(conf)


  /**
   * RDD[T]抽象类的部分相关代码   注意下: 这些方法的权限修饰符都是protected
   * -
   * 获取所有分区
   * protected def getPartitions: Array[Partition]
   * -
   * 获取所有依赖关系
   * protected def getDependencies: Seq[Dependency[_]] = deps
   * -
   * 获取优先位置列表
   * protected def getPreferredLocations(split: Partition): Seq[String] = Nil
   * -
   * 分区器 有子类重写以及指定他们的分区方式
   *
   * @transient val partitioner: Option[Partitioner] = None
   */
  @Test
  def rdd_T(): Unit = {
    // 1.数据生成, 默认分区数为程序所分配到的CPU的核心数
    val data = Array(1, 2, 3, 4, 5)
    val numData: RDD[Int] = sc.parallelize(data)
    // 2.查看分区数
    /**
     * numData.getNumPartitions = 6
     * 得到6的原因, 是因为SparkConf().setMaster("local[6]"), 设置了6个本地线程
     */
    println(s"numData.getNumPartitions = ${numData.getNumPartitions}")

    val numData2 = sc.parallelize(data, 3)
    // numData2.getNumPartitions = 3
    println(s"numData2.getNumPartitions = ${numData2.getNumPartitions}")

  }

  /**
   * mapPartitions 和 map 算子是一样的, 只不过 map 是针对每一条数据进行转换,
   * .             而 mapPartitions 是针对一整个分区的数据进行转换
   * 所以:
   * 1. map 的 func 参数是单条数据, mapPartitions 的 func 参数是一个集合(一个分区整个所有的数据)
   * 2. map 的 func 返回值也是单条数据, mapPartitions 的 func 返回值是一个集合
   * 函数原型
   * def mapPartitions[U: ClassTag](
   * .    f: Iterator[T] => Iterator[U],
   * .    preservesPartitioning: Boolean = false): RDD[U]
   * ##
   * preservesPartitioning: Boolean = false
   * 这是scala定义方法的中的, 默认参数, 调用方法时没有传入这个参数, 将把默认值传给方法
   */
  @Test
  def mapPartitions(): Unit = {
    // 1.数据生成
    val numData = sc.parallelize(Seq(1, 2, 3, 4, 5, 6), 2)

    numData.mapPartitions(iter => {
      // 这个函数必须返回值, 所以使用语法块表达式, 返回值是{}内部最后一行的值
      iter.foreach(println(_))
      iter
    }).collect()
    // 2.算子使用

  }

  @Test
  def mapPartitions2(): Unit = {
    // 1.数据生成
    val numData = sc.parallelize(Seq(1, 2, 3, 4, 5, 6), 3)

    // 2.算子使用

    // 想把集合中的数据每条加10
    numData.mapPartitions((iter: Iterator[Int]) => {
      // 遍历iter其中每一条数据进行转换, 转换完成后, 返回这个iter
      //  iter 是 scala 中的集合类型, 且iter的map方法是使用scala内置的, 并不是RDD算子
      iter.map(x => x + 10)
    }).collect()
      .foreach(println(_))
  }

  /**
   * mapPartitionsWithIndex 和 mapPartitions 的区别是 func 中多了一个参数, 是分区号
   * 函数原型
   * def mapPartitionsWithIndex[U: ClassTag](
   * .    f: (Int, Iterator[T]) => Iterator[U],
   * .    preservesPartitioning: Boolean = false): RDD[U]
   */
  @Test
  def mapPartitionsWithIndex(): Unit = {
    // 1.数据生成
    val numData = sc.parallelize(Seq(1, 2, 3, 4, 5, 6), 3)

    // 2.算子使用
    numData.mapPartitionsWithIndex((index: Int, iter: Iterator[Int]) => {
      println(s"index = ${index}")
      iter.foreach(x => println(s"${index} 分区:  ${x}"))
      iter
    }).collect()

    /**
     * numSlices = 2      2               3
     * index = 0      index = 1       index = 0
     * index = 1      index = 0       index = 2
     * 1              1               index = 1
     * 4              4               0 分区:  1
     * 2              5               1 分区:  3
     * 5              6               2 分区:  5
     * 3              2               1 分区:  4
     * 6              3               0 分区:  2
     * .                              2 分区:  6
     */

    numData.mapPartitionsWithIndex((index: Int, iter: Iterator[Int]) => {
      val buffer = new ListBuffer[String]
      while (iter.hasNext) {
        buffer.append(index + " 分区: " + iter.next() * 100)
      }
      buffer.toIterator
    }).collect()
      .foreach(println(_))

    /**
     * 0 分区: 100
     * 0 分区: 200
     * 1 分区: 300
     * 1 分区: 400
     * 2 分区: 500
     * 2 分区: 600
     */
  }


  /**
   * filter 可以过滤掉数据中一个部分元素
   * filter 中接收的函数, 参数是每一个元素, 如果这个函数返回true,
   * .      当前元素就会被加入到数据集, 否则当前元素就会被过滤掉
   */
  @Test
  def filter(): Unit = {
    // 1.定义集合
    val numData = sc.parallelize(Seq(-5, -4, -3, -2, -1, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10), 1)
    // 2.过滤数据
    // 含有 正整数和负整数中的集合, 过滤出偶数, 需要使用函数 x % 2 == 0
    //                           过滤出奇数, 需要使用函数 x % 2 != 0
    numData.filter(x => x % 2 != 0)
      .collect()
      .foreach(println(_))
    // 3.收集数据
  }


  /**
   * 作用: 把大数据集变小, 尽可能的减少数据集规律的损失
   * withReplacement: 指定为true的情况下[取样后还放回原数据集供下次使用], 采集到的数据会出现重复
   * .                指定为false的情况下[并没有放回去, 符合正常现象], 数据不重复
   * #
   * 函数原型
   * def sample(
   * .    withReplacement: Boolean,
   * .    fraction: Double,               //  意为抽样的比例
   * .    seed: Long = Utils.random.nextLong): RDD[T]
   */
  @Test
  def sample(): Unit = {
    // 1.定义集合
    val numData = sc.parallelize(Seq(-5, -4, -3, -2, -1, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10))
    // 2.过滤数据
    val result = numData.sample(withReplacement = false, 0.6)
      .collect()
    // 3.收集结果
    result.foreach(println(_))
  }


  /**
   * mapValue也是map, 只不过map作用于整条数据, mapValue作用于Value
   * 函数原型
   * def mapValues[U](f: V => U): RDD[(K, U)]
   */
  @Test
  def mapValues(): Unit = {
    val mapData = sc.parallelize(Seq(("hadoop", 1), ("hive", 2), ("spark", 3)))

    // 下面这个两种方式相同, 只是mapValue更加简便了
    mapData.map((x: (String, Int)) => x._1 -> x._2 * 10).collect().foreach(println(_))
    println("-" * 60)
    mapData.mapValues((x: Int) => x * 10).collect().foreach(println(_))
  }


  /**
   * intersection 交集
   * 函数原型
   * def intersection(other: RDD[T]): RDD[T]
   */
  @Test
  def intersection(): Unit = {
    val rdd1 = sc.parallelize(Seq(1, 2, 3, 4, 5, 6))
    val rdd2 = sc.parallelize(Seq(3, 4, 5, 6, 7))

    rdd1.intersection(rdd2)
      .collect()
      .foreach(println(_))
  }


  /**
   * union 并集  -- 会出现元素重复  * 可以使用`.distinct（）`消除重复元素
   * 函数原型
   * def union(other: RDD[T]): RDD[T]
   * #
   * distinct 是一个需要shuffled的操作, 本质上distinct就是一个reduceByKey, 把重复的合并为一个
   */
  @Test
  def union(): Unit = {
    val rdd1 = sc.parallelize(Seq(1, 2, 3, 4, 5, 6))
    val rdd2 = sc.parallelize(Seq(3, 4, 5, 6, 7))

    rdd1.union(rdd2)
      .distinct() // 使用 distinct() 消除rdd中的重复元素
      .collect()
      .foreach(println(_))

  }


  /**
   * subtract 差集  ->  A.subtract(B)  == 集合A - 集合A 交 集合B
   * 函数原型
   * def subtract(other: RDD[T]): RDD[T]
   */
  @Test
  def subtract(): Unit = {
    val rdd1 = sc.parallelize(Seq(1, 2, 3, 4, 5, 6))
    val rdd2 = sc.parallelize(Seq(3, 4, 5, 6, 7))

    rdd1.subtract(rdd2)
      .collect()
      .foreach(println(_))

  }


  /**
   * groupByKey算子的主要作用是按照key分组, 和reduceByKey很类似
   * 但是groupByKey并不求聚合, 只是列举key对应的所有value
   * #
   * groupByKey运算结果的格式: (K, (value1, value2))
   * reduceByKey能不能在Map端做Combiner : 1 能不能减少IO
   * groupByKey在map端做Combiner有没有意义 ? 没有的... ???
   * 函数原型
   * def groupByKey(): RDD[(K, Iterable[V])]
   */
  @Test
  def groupByKey(): Unit = {
    val mapData = sc.parallelize(Seq(("hadoop", 1), ("hive", 2), ("spark", 3), ("hadoop", 4)))

    mapData.groupByKey()
      .collect()
      .foreach(println(_))

    /**
     * 输出结果
     * (hadoop,CompactBuffer(1, 4))
     * (hive,CompactBuffer(2))
     * (spark,CompactBuffer(3))
     */
  }


  /**
   * CombineByKey 这个算子中接收三个函数
   * 转换数据的函数(初始函数, 作用域第一条数据, 用于开启整个计算), 在分区上的结果再次聚合, 生成最终结果
   * 函数原型
   * def combineByKey[C](
   * .    createCombiner: V => C,
   * .    mergeValue: (C, V) => C,
   * .    mergeCombiners: (C, C) => C,
   * .    partitioner: Partitioner,
   * .    mapSideCombine: Boolean = true,
   * .    serializer: Serializer = null): RDD[(K, C)]
   */
  @Test
  def combinerByKey(): Unit = {
    // 1.准备集合
    val rdd = sc.parallelize(Seq(
      ("zhangsan", 99.0),
      ("zhangsan", 96.0),
      ("lisi", 97.0),
      ("lisi", 98.0),
      ("zhangsan", 97.0)
    ))

    // 2.算子操作
    //    2.1.createCombiner 转换数据
    //    2.2.mergeValue 分区上的聚合
    //    2.3.mergeCombiners 把所有分区上的结果再次聚合, 生成最终结果
    val result = rdd.combineByKey(
      createCombiner = (curr: Double) => (curr, 1),
      mergeValue = (curr: (Double, Int), nextValue: Double) => (curr._1 + nextValue, curr._2 + 1),
      mergeCombiners = (curr: (Double, Int), agg: (Double, Int)) => (curr._1 + agg._1, curr._2 + agg._2))


    // ("zhangsan", (99 + 96 + 97, 3))

    // 3.获取结果, 打印结果
    result.map((item: (String, (Double, Int))) => item._1 -> item._2._1 / item._2._2)
      .collect()
      .foreach(println(_))

    /**
     * (zhangsan,97.33333333333333)
     * (lisi,97.5)
     */
  }


  /**
   * foldByKey和Spark中的reduceByKey的区别是指定初始值
   * foldByKey和Scala中foldLeft或者foldRight区别是, 这个初始值作用于每一个数据
   * .            而Scala中的foldLeft是给一个数据集指定一个初始值
   * 函数原型
   * def foldByKey(zeroValue: V)(func: (V, V) => V): RDD[(K, V)]
   */
  @Test
  def foldByKey(): Unit = {
    val mapData = sc.parallelize(Seq(("hadoop", 1), ("hive", 2), ("spark", 3), ("hadoop", 4)))

    mapData.foldByKey(10)((total: Int, curr: Int) => curr + total)
      .collect()
      .foreach(println(_))

    /**
     * (hadoop,25)
     * (hive,12)
     * (spark,13)
     */
  }


  /**
   * 作用: 集合所有key相同的value, 按照key聚合value
   * #
   * aggregateByKey(zeroValue)(seqOp, combOp)
   * zeroValue: 指定初始值
   * seqOp: 作用于每一个元素, 根据初始值, 进行计算
   * combOp: 将seqOp处理过的结果进行聚合
   * -
   * aggregateByKey 特别适合针对每个数据要先处理, 再聚合
   * 函数原型
   * def aggregateByKey[U: ClassTag](zeroValue: U)(seqOp: (U, V) => U,
   * .      combOp: (U, U) => U): RDD[(K, U)]
   * #
   * aggregateByKey和reduceByKey的区别:
   * aggregateByKey最终聚合结果的类型和传入的初始值[zeroValue]类型保持一致
   * reduceByKey在集合中选取第一个值作为初始值, 并且聚合过的数据类型不能改变
   */
  @Test
  def aggregateByKey(): Unit = {
    // 需求 : 每个商品都要打8折, 然后再计算每种商品的打折后的总价值
    val rdd = sc.parallelize(Seq(("手机", 10.0), ("手机", 15.0), ("电脑", 20.0)))

    rdd.aggregateByKey(0.8)(
      seqOp = (zeroValue: Double, price: Double) => price * zeroValue,
      combOp = (total, curr) => total + curr)
      .collect()
      .foreach(println(_))

    /**
     * (手机,20.0)
     * (电脑,16.0)
     */

    rdd.reduceByKey((total: Double, curr: Double) => total + curr)
      .collect()
      .foreach(println(_))

    /**
     * (手机,25.0)
     * (电脑,20.0)
     */
  }


  /**
   * 作用:
   * .    将两个RDD按照相同的key进行连接
   * 函数原型:
   * .    def join[W](other: RDD[(K, W)]): RDD[(K, (V, W))]
   * .    def join[W](other: RDD[(K, W)], numPartitions: Int): RDD[(K, (V, W))]
   * 参数:
   * .    other其他RDD
   * .    partitioner 或者 numPartitions 可选, 可以通过传递分区函数或者分区数量来改变分区
   * 注意点:
   * .    join有点类似于sql中的内连接, 只会在结果中包含能够连接到的key
   * .    join的结果是一个笛卡尔积形式
   * 如果想执行外连接, 可以使用leftOuterJoin, rightOuterJoin和fullOuterJoin
   */
  @Test
  def join(): Unit = {
    val rdd1 = sc.parallelize(Seq(("a", 1), ("a", 2), ("b", 1)))
    val rdd2 = sc.parallelize(Seq(("a", 10), ("a", 11), ("a", 12)))

    rdd1.join(rdd2)
      .collect()
      .foreach(println(_))

    /**
     * (a,(1,10))
     * (a,(1,11))
     * (a,(1,12))
     * (a,(2,10))
     * (a,(2,11))
     * (a,(2,12))
     */


  }

  /**
   * 函数原型
   * def leftOuterJoin[W](other: RDD[(K, W)]): RDD[(K, (V, Option[W]))]
   * #
   * def rightOuterJoin[W](other: RDD[(K, W)]): RDD[(K, (Option[V], W))]
   */
  @Test
  def leftOuterJoin(): Unit = {
    val rdd1 = sc.parallelize(Seq(("a", 1), ("a", 2), ("b", 1)))
    val rdd2 = sc.parallelize(Seq(("a", 10), ("a", 11), ("a", 12), ("d", 20)))
    rdd1.leftOuterJoin(rdd2)
      .collect()
      .foreach(println(_))

    /**
     * (a,(1,Some(10)))
     * (a,(1,Some(11)))
     * (a,(1,Some(12)))
     * (a,(2,Some(10)))
     * (a,(2,Some(11)))
     * (a,(2,Some(12)))
     * (b,(1,None))
     */

    rdd1.rightOuterJoin(rdd2)
      .collect()
      .foreach(println(_))

    /**
     * (a,(Some(1),10))
     * (a,(Some(1),11))
     * (a,(Some(1),12))
     * (a,(Some(2),10))
     * (a,(Some(2),11))
     * (a,(Some(2),12))
     * (d,(None,20))
     */
  }


  /**
   * 计算笛卡尔积
   * 函数原型
   * def cartesian[U: ClassTag](other: RDD[U]): RDD[(T, U)]
   */
  @Test
  def cartesian(): Unit = {
    val rdd1 = sc.parallelize(Seq("A", "B", "C"))
    val rdd2 = sc.parallelize(Seq(1, 2, 3))

    rdd1.cartesian(rdd2)
      .collect()
      .foreach(println(_))

    /**
     * (A,1)    (B,1)     (C,1)
     * (A,2)    (B,2)     (C,2)
     * (A,3)    (B,3)     (C,3)
     */
  }


  /**
   * sortBy 可以作用任何类型数据的RDD, sortByKey只有KV类型数据的RDD中才有
   * sortBy 可以按照任何部分来排序, sortByKey只能按照key来排序
   * sortByKey 写法简单, 不用编写函数
   *
   * sortBy 函数原型
   * def sortBy[K](
   * .    f: (T) => K,
   * .    ascending: Boolean = true,
   * .    numPartitions: Int = this.partitions.length)
   * .    (implicit ord: Ordering[K], ctag: ClassTag[K]): RDD[T]
   * #
   * sortByKey 函数原型
   * def sortByKey(ascending: Boolean = true,
   * .            numPartitions: Int = self.partitions.length): RDD[(K, V)]
   */
  @Test
  def sort(): Unit = {
    val numData = sc.parallelize(Seq(2, 4, 1, 5, 7, 9, 0))
    val mapData = sc.parallelize(Seq(("a", 1), ("c", 2), ("b", 4)))

    // 0 1 2 4 5 7 9
    numData.sortBy((item: Int) => item).collect().foreach(x => print(x + " "))
    println("\n" + "-" * 60)

    // (a,1) (c,2) (b,4)
    mapData.sortBy((item: (String, Int)) => item._2).collect().foreach(x => print(x + " "))
    println("\n" + "-" * 60)

    // (a,1) (b,4) (c,2)
    mapData.sortByKey().collect().foreach(x => print(x + " "))
  }

  /**
   * 控制分区操作 reparation 和 coalesce
   * #
   * repartition进行重分区的时候, 默认是shuffle的, 默认可以整大和缩小分区数
   * coalesce进行重分区的时候, 默认不是shuffle, coalesce默认不能增大分区数
   * 函数原型
   * def repartition(numPartitions: Int)(implicit ord: Ordering[T] = null): RDD[T]
   * #
   * def coalesce(numPartitions: Int, shuffle: Boolean = false,
   * .            partitionCoalescer: Option[PartitionCoalescer] = Option.empty)
   * .            (implicit ord: Ordering[T] = null)
   * .          : RDD[T]
   */
  @Test
  def partitioning(): Unit = {
    var rdd = sc.parallelize(Seq(1, 2, 3, 4, 5, 6), 2)

    // repartition  可大可小
    //println(rdd.repartition(5).partitions.size)   // 5
    //println(rdd.repartition(1).partitions.size)   // 1

    // coalesce
    println(rdd.coalesce(5, shuffle = true).partitions.size) // 2
  }
}